package com.example.mq.mqserver;

import com.example.mq.common.arguments.*;
import com.example.mq.common.dto.BasicReturns;
import com.example.mq.common.dto.Request;
import com.example.mq.common.dto.Response;
import com.example.mq.common.dto.SubScribeReturns;
import com.example.mq.common.exception.MqException;
import com.example.mq.common.tool.BinaryTool;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.*;

/*
 * 这个 BrokerServer 就是咱们 消息队列 本体服务器.
 * 本质上就是一个 TCP 的服务器.
 */
@Slf4j
public class BrokerServer {
    private ServerSocket serverSocket = null;

    // 当前考虑一个 BrokerServer 上只有一个 虚拟主机
    private final VirtualHost virtualHost = new VirtualHost("default");
    // 使用这个 哈希表 表示当前的所有会话(也就是说有哪些客户端正在和咱们的服务器进行通信)
    // 此处的 key 是 channelId, value 为对应的 Socket 对象
    private final ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<String, Socket>();
    // 引入一个线程池, 来处理多个客户端的请求.
    private ExecutorService executorService = null;
    // 引入一个 boolean 变量控制服务器是否继续运行
    private volatile boolean runnable = true;

    // BrokerServer 构造函数，接受一个端口号作为参数，用于初始化服务器套接字
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port); // 在指定端口上创建一个服务器套接字
    }

    // start 方法用于启动 BrokerServer
    public void start() throws IOException {
        log.info("[BrokerServer] 启动!"); // 记录服务器启动的日志信息

        // 创建一个缓存的线程池，用于处理客户端连接
        executorService = new ThreadPoolExecutor(
                4, // 核心线程数
                8, // 最大线程数
                30L, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(50), // 阻塞队列
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );

        try {
            while (runnable) { // 服务器运行的主循环，runnable 控制服务器是否继续运行
                Socket clientSocket = serverSocket.accept(); // 接受客户端连接
                // 将处理客户端连接的逻辑提交给线程池，线程池会为每个客户端连接分配一个线程
                executorService.submit(() -> {
                    processConnection(clientSocket); // 处理客户端连接
                });
            }
        } catch (SocketException e) {
            log.error("[BrokerServer] 服务器停止运行!"); // 记录服务器停止运行的日志信息
            log.error(e.getMessage()); // 记录异常的详细信息
        }
    }


    // 一般来说停止服务器, 就是直接 kill 掉对应进程就行了.
    // 此处还是搞一个单独的停止方法. 主要是用于后续的单元测试.
    public void stop() throws IOException {
        runnable = false;
        // 把线程池中的任务都放弃了. 让线程都销毁.
        executorService.shutdownNow();
        serverSocket.close();
    }

    // 通过这个方法, 来处理一个客户端的连接.
    // 在这一个连接中, 可能会涉及到多个请求和响应.
    private void processConnection(Socket clientSocket) {

        try (DataInputStream dataInputStream = new DataInputStream(clientSocket.getInputStream());
             DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream())) {
            while (true) {
                // 1. 读取请求并解析.
                Request request = readRequest(dataInputStream);
                // 2. 根据请求计算响应
                Response response = process(request, clientSocket);
                // 3. 把响应写回给客户端
                writeResponse(dataOutputStream, response);
            }
        } catch (EOFException | SocketException e) {
            log.info("[BrokerServer] connection 关闭! 客户端的地址: {}:{}", clientSocket.getInetAddress().toString(), clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e) {
            // 处理其他异常，并记录详细日志
            log.error("[BrokerServer] connection 出现异常!", e);
        } finally {
            // 关闭客户端Socket连接并清理会话
            try {
                clientSocket.close(); // 关闭Socket连接
                clearClosedSession(clientSocket); // 清理该Socket对应的会话
            } catch (IOException e) {
                log.error("关闭客户端Socket时出错", e); // 记录关闭Socket时的异常
            }
        }
    }

    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();

        // 读取请求类型
        request.setType(dataInputStream.readInt());

        // 读取请求的长度
        int length = dataInputStream.readInt();
        request.setLength(length);

        // 读取请求的负载数据
        byte[] payload = new byte[length];
        int bytesRead = dataInputStream.read(payload);

        // 检查是否读取到了预期数量的数据
        if (bytesRead != length) {
            throw new IOException("读取请求格式出错! 期望读取 " + length + " 字节, 实际读取 " + bytesRead + " 字节");
        }

        request.setPayload(payload);
        return request;
    }


    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 这个刷新缓冲区也是重要的操作!!
        dataOutputStream.flush();
    }

    private Response process(Request request, Socket clientSocket)
            throws IOException, ClassNotFoundException, MqException {
        // 日志：开始处理请求
        log.info("[BrokerServer] 开始处理请求, type={}, channelId={}, rid={}",
                request.getType(), request.getLength(), clientSocket.getInetAddress().toString()
                        + ":" + clientSocket.getPort());

        // 1. 把 request 中的 payload 做一个初步的解析.
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        log.info("[Request] rid={}, channelId={}, type={}, length={}",
                basicArguments.getRid(), basicArguments.getChannelId(), request.getType(), request.getLength());

        boolean ok = true;

        // 2. 根据 request 的 type 值，使用 switch 语句区分请求类型
        switch (request.getType()) {
            case 0x1: // 创建 channel
                sessions.put(basicArguments.getChannelId(), clientSocket);
                log.info("[BrokerServer] 创建 channel 完成! channelId={}", basicArguments.getChannelId());
                break;
            case 0x2: // 销毁 channel
                sessions.remove(basicArguments.getChannelId());
                log.info("[BrokerServer] 销毁 channel 完成! channelId={}", basicArguments.getChannelId());
                break;
            case 0x3: // 创建交换机
                ExchangeDeclareArguments exchangeDeclareArgs = (ExchangeDeclareArguments) basicArguments;
                ok = virtualHost.exchangeDeclare(exchangeDeclareArgs.getExchangeName(), // 交换机名
                        exchangeDeclareArgs.getExchangeType(),                          // 交换机类型
                        exchangeDeclareArgs.isDurable(),                                // 是否持久化
                        exchangeDeclareArgs.isAutoDelete(),                             // 是否自动闲置删除
                        exchangeDeclareArgs.getArguments());                            // 交换机参数
                log.info("[BrokerServer] 创建交换机, exchangeName={}, type={}, result={}",
                        exchangeDeclareArgs.getExchangeName(), exchangeDeclareArgs.getExchangeType(), ok);
                break;
            case 0x4: // 删除交换机
                ExchangeDeleteArguments exchangeDeleteArgs = (ExchangeDeleteArguments) basicArguments;
                ok = virtualHost.exchangeDelete(exchangeDeleteArgs.getExchangeName());
                log.info("[BrokerServer] 删除交换机, exchangeName={}, result={}", exchangeDeleteArgs.getExchangeName(), ok);
                break;
            case 0x5: // 创建队列
                QueueDeclareArguments queueDeclareArgs = (QueueDeclareArguments) basicArguments;
                ok = virtualHost.queueDeclare(queueDeclareArgs.getQueueName(), queueDeclareArgs.isDurable(),
                        queueDeclareArgs.isExclusive(),     // 设置是否独占
                        queueDeclareArgs.isAutoDelete(),    // 是否自动闲置删除
                        queueDeclareArgs.getArguments());   // 设置参数
                log.info("[BrokerServer] 创建队列, queueName={}, result={}", queueDeclareArgs.getQueueName(), ok);
                break;
            case 0x6: // 删除队列
                QueueDeleteArguments queueDeleteArgs = (QueueDeleteArguments) basicArguments;
                ok = virtualHost.queueDelete(queueDeleteArgs.getQueueName());
                log.info("[BrokerServer] 删除队列, queueName={}, result={}",
                        queueDeleteArgs.getQueueName(), ok);
                break;
            case 0x7: // 绑定队列
                QueueBindArguments queueBindArgs = (QueueBindArguments) basicArguments;
                ok = virtualHost.queueBind(queueBindArgs.getQueueName(), queueBindArgs.getExchangeName(),
                        queueBindArgs.getBindingKey());
                log.info("[BrokerServer] 绑定队列, queueName={}, exchangeName={}, bindingKey={}, result={}",
                        queueBindArgs.getQueueName(), queueBindArgs.getExchangeName(),
                        queueBindArgs.getBindingKey(), ok);
                break;
            case 0x8: // 解绑队列
                QueueUnbindArguments queueUnbindArgs = (QueueUnbindArguments) basicArguments;
                ok = virtualHost.queueUnbind(queueUnbindArgs.getQueueName(), queueUnbindArgs.getExchangeName());
                log.info("[BrokerServer] 解绑队列, queueName={}, exchangeName={}, result={}",
                        queueUnbindArgs.getQueueName(), queueUnbindArgs.getExchangeName(), ok);
                break;
            case 0x9: // 发布消息
                BasicPublishArguments basicPublishArgs = (BasicPublishArguments) basicArguments;
                ok = virtualHost.basicPublish(basicPublishArgs.getExchangeName(),   // 交换机名
                        basicPublishArgs.getRoutingKey(),   // key
                        basicPublishArgs.getBasicProperties(),  // 消息属性
                        basicPublishArgs.getBody());    // 内容
                log.info("[BrokerServer] 发布消息, exchangeName={}, routingKey={}, result={}",
                        basicPublishArgs.getExchangeName(), basicPublishArgs.getRoutingKey(), ok);
                break;
            case 0xa: // 消费消息
                BasicConsumeArguments basicConsumeArgs = (BasicConsumeArguments) basicArguments;
                ok = virtualHost.basicConsume(basicConsumeArgs.getConsumerTag(), basicConsumeArgs.getQueueName(),
                        basicConsumeArgs.isAutoAck(),
                        (consumerTag, basicProperties, body) -> {
                            // 获取对应的客户端 Socket
                            Socket clientSocket1 = sessions.get(consumerTag);
                            if (clientSocket1 == null || clientSocket1.isClosed()) {
                                throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
                            }
                            // 构造响应数据
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            subScribeReturns.setChannelId(consumerTag);
                            subScribeReturns.setRid(""); // 由于这里只有响应, 没有请求, 不需要去对应. rid 暂时不需要.
                            subScribeReturns.setOk(true);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            byte[] payload = BinaryTool.toBytes(subScribeReturns);
                            Response response = new Response();
                            response.setType(0xc); // 服务器给消费者客户端推送的消息数据.
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            // 把数据写回给客户端
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket1.getOutputStream());
                            writeResponse(dataOutputStream, response);
                            log.info("[BrokerServer] 向客户端推送消息, consumerTag={}, length={}",
                                    consumerTag, payload.length);
                        });
                log.info("[BrokerServer] 消费消息, consumerTag={}, queueName={}, result={}",
                        basicConsumeArgs.getConsumerTag(), basicConsumeArgs.getQueueName(), ok);
                break;
            case 0xb: // 确认消息
                BasicAckArguments basicAckArgs = (BasicAckArguments) basicArguments;
                ok = virtualHost.basicAck(basicAckArgs.getQueueName(), basicAckArgs.getMessageId());
                log.info("[BrokerServer] 确认消息, queueName={}, messageId={}, result={}",
                        basicAckArgs.getQueueName(), basicAckArgs.getMessageId(), ok);
                break;
            default: // 未知的 type
                log.error("[BrokerServer] 未知的 type! type={}", request.getType());
                throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
        }

        // 3. 构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toBytes(basicReturns);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);

        // 日志：处理完毕，构造响应
        log.info("[Response] rid={}, channelId={}, type={}, length={}",
                basicReturns.getRid(), basicReturns.getChannelId(), response.getType(), response.getLength());

        return response;
    }


    private void clearClosedSession(Socket clientSocket) {
        // 使用 removeIf 直接在遍历时删除匹配的元素
        sessions.entrySet().removeIf(entry -> {
            boolean shouldRemove = entry.getValue() == clientSocket;
            if (shouldRemove) {
                log.info("[BrokerServer] 清理 session 完成! 被清理的 channelId={}", entry.getKey());
            }
            return shouldRemove;
        });
    }
}
